package com.xy.job.admin.core.schedule

import com.xy.job.admin.core.ExecutorClient
import com.xy.job.admin.domain.JobInstance
import com.xy.job.admin.domain.NodeRegistry
import com.xy.job.admin.enums.NodeType
import com.xy.job.admin.repository.JobInstanceRepository
import com.xy.job.admin.repository.NodeRegistryRepository
import com.xy.job.core.api.request.JobTriggerParam
import com.xy.job.core.enums.JobStatus
import com.xy.job.core.exception.BizException
import com.xy.job.core.utils.getIP
import org.slf4j.LoggerFactory
import java.time.LocalDateTime

class TriggerTask(
    private val jobInstance: JobInstance,
    private val nodeRegistryRepository: NodeRegistryRepository,
    private val jobInstanceRepository: JobInstanceRepository,
) : Runnable {

    private val log = LoggerFactory.getLogger(TriggerTask::class.java)

    override fun run() {
        val executorRegistries = nodeRegistryRepository.findByNodeNameAndNodeType(
            nodeName = jobInstance.groupCode!!,
            nodeType = NodeType.EXECUTOR,
        )
        jobInstance.scheduleTime = LocalDateTime.now()
        jobInstance.schedulerIp = getIP()
        if (executorRegistries.isEmpty()) {
            jobInstance.jobStatus = JobStatus.FAILED
            jobInstance.completeTime = LocalDateTime.now()
            jobInstance.message = "无可用机器"
        } else {
            trigger(executorRegistries.shuffled())
        }
        jobInstanceRepository.saveAndFlush(jobInstance)
    }

    private fun trigger(executorRegistries: List<NodeRegistry>) {
        val jobTriggerParam = JobTriggerParam(
            jobId = jobInstance.jobId!!,
            jobInstanceId = jobInstance.id!!,
            jobName = jobInstance.jobName!!,
            jobHandlerName = jobInstance.jobHandler!!,
            executeParams = jobInstance.executeParams,
            timeout = (jobInstance.extendConfig?.timeoutPeriod ?: 0) * 1000L,
            dataTime = jobInstance.dataTime!!,
            scriptType = jobInstance.scriptType,
            scriptSource = jobInstance.scriptCode,
        )

        for (executorRegistry in executorRegistries) {
            try {
                jobInstance.executorAddress = executorRegistry.nodeAddress
                val executorClient = ExecutorClient(
                    host = executorRegistry.nodeHost!!,
                    port = executorRegistry.nodePort!!,
                )
                val result = executorClient.trigger(jobTriggerParam)
                if (result.success) {
                    log.info("job trigger success, jobInstanceId={}", jobInstance.id)
                    jobInstance.jobStatus = JobStatus.PROCESSING
                } else {
                    log.info("job trigger fail, jobInstanceId={}", jobInstance.id)
                    jobInstance.jobStatus = JobStatus.FAILED
                    jobInstance.message = result.message
                }
            } catch (e: Throwable) {
                log.error(
                    "job trigger failed, jobInstanceId={}, jobName={}, executorAddr={}, message={}",
                    jobInstance.id, jobInstance.jobName,
                    jobInstance.executorAddress, e.message, e
                )
                jobInstance.jobStatus = JobStatus.FAILED
                jobInstance.message = when (e) {
                    is BizException -> e.message
                    else -> e.stackTraceToString()
                }
            }
        }
    }
}